package com.ass.rocketmq.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.ass.rocketmq.constant.MQTopic;
import com.ass.rocketmq.producer.DefaultProducer;
import com.domain.MQError;
import lombok.SneakyThrows;
import lombok.extern.log4j.Log4j2;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationPreparedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.Objects;
import java.util.UUID;

@Slf4j
//让所有的继承该类的消费者 都传个泛型过来 有实体的传实体 没实体的传别的也行 例如字符串
//限制子类直接让其定义好泛型 该用什么类型接 跟据业务来
//这个M就是个Object的类型
@Component
//并发异步消费 旨在提升消费速度 不关注消费顺序
public abstract class AbstractConcurrencyPushConsumer<M>  {


    @Value("${spring.profiles.active}")
    private String active;

    @Value("${rocketmq.accessKey}")
    private String accessKey;

    @Value("${rocketmq.secretKey}")
    private String secretKey;

    @Value("${rocketmq.namespace}")
    private String namespace;


    @Value("${rocketmq.name-server}")
    private String namesrvAddr;


//    private Class<M> mClass;

    //解决泛型擦除
    private TypeReference<M> typeReference;

    @Autowired
    private DefaultProducer errorProducer;


    /**
     * 初始化mq消费者
     */
    public DefaultMQPushConsumer initConsumer() {
        String prefix = active + "_";
        MQTopic mqTopic = getMqTopic();
        //如果用户名密码不对 消费者消费不到消息
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(namespace,prefix + mqTopic.tag,
                new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)));
        //消息息队列分配策略 这个策略和集群模式消费有冲突 使用时一定要注意
//        consumer.setAllocateMessageQueueStrategy(new AbstractAllocateMessageQueueStrategy());
        //设置每次批量消费的消息数量 批量方式消费  看情况打开，开启后 就必须考虑幂等问题了 在消费端
//        consumer.setConsumeMessageBatchMaxSize(10);
        //设置最大重试次数
//        consumer.setMaxReconsumeTimes(1);
        /**设置采用集群模式，
         * RocketMQ Push Consumer默认为集群模式，同一个消费组内的消费者分担消费
         *1.在集群模式下，多个消费者实例共同消费同一个主题的消息。
         *2.消息被发送到一个主题，然后由消费者组中的每个消费者实例中的一个进行消费。
         *3.每个消费者实例只消费主题的一个子集，即消息被分区（tag）到各个消费者实例中。
         *4.集群模式适用于需要多个消费者实例共同处理消息的场景，以提高消息的处理能力和容错性。
          5.用于解决主题阻塞问题 (生产者的发送消息的速度远远大于消费者消费的消息的速度称之为阻塞)

         * 广播模式:consumer.setMessageModel(MessageModel.BROADCASTING)，消费组内的每一个消费者都会消费全量消息。
         * 1.在广播模式下，每个消费者实例都会独立消费同一个主题的所有消息。
         * 2.每个消费者实例都会接收到主题的所有消息副本，而不是仅接收主题的一个子集。
         * 3.广播模式适用于需要每个消费者实例都独立处理所有消息的场景，例如日志收集、实时统计等。
         */
        consumer.setMessageModel(MessageModel.CLUSTERING);
        try {
            consumer.subscribe(prefix + mqTopic.topic, prefix + mqTopic.tag);
        } catch (MQClientException e) {
            log.error("[消费者初始化异常]", e);
        }
        //并发异步消费：MessageListenerConcurrently
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            log.info("消息的大小：{}",msgs.size());
            for (MessageExt msg : msgs) {
                try {
                    /**RocketMQ无法避免消息重复（Exactly-Once），
                      所以如果业务对消费重复非常敏感，务必要在业务层面进行去重处理。
                      可以借助关系数据库进行去重。首先需要确定消息的唯一键，可以是msgId，也可以是消息内容中的唯一标识字段，
                      例如订单Id等。在消费之前判断唯一键是否在关系数据库中存在。如果不存在则插入，并消费，否则跳过。
                     （实际过程要考虑原子性问题，判断是否存在可以尝试插入，如果报主键冲突，则插入失败，直接跳过）*/
                    handlerMsg(msg);
                } catch (Exception e) {
                    //阿里二面的面试题答案
                    //如果一批消息按照顺序消费，是不可能出现第 100 条消息消费成功了，但第 50 条消费失败的情况，
                    // 因为第 50 条消息失败的时候，应该退出循环，不再继续进行消费。
                    //如果是并发消费，如果出现了这种情况，建议是整批消息全部重新消费，也就是给 ackIndex 赋值 0，
                    // 这样必须考虑幂等问题。
                    e.printStackTrace();
                    //这是对框架的重试 而不是业务的重试
                    if(msg.getReconsumeTimes()<=1){
                        //这个得放这 要不会重试很多次 rocketmq 自己的重试机制
                        //结束循环 让这一批次的消息都进行重试
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }else {
                        //有个缺点无法将重试达到次数的消息发送到死信队列中 因为最后的CONSUME_SUCCESS 会认为消息消费成功了
                        log.error("消费异常重试次数达到上线：{}",msg.getReconsumeTimes());
                        //直接存储到记录的日志
                        errorConsumerCollect(mqTopic,
                                new String(msg.getBody(), StandardCharsets.UTF_8),e);
                        //收集完mq异常日志信息后跳出循环
                        break;
                    }
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            try {
//                有个缺点 可以将消息发送到死信队列种 但是死信队列的topic的权限是2 得手动改成6 消费者才可以消费到
//                for (MessageExt msg : msgs) {
//                    handlerMsg(msg);
//                }
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            } catch (Exception e) {
//                //阿里二面的面试题答案
//                //如果一批消息按照顺序消费，是不可能出现第 100 条消息消费成功了，但第 50 条消费失败的情况，
//                // 因为第 50 条消息失败的时候，应该退出循环，不再继续进行消费。
//                //如果是并发消费，如果出现了这种情况，建议是整批消息全部重新消费，也就是给 ackIndex 赋值 0，
//                // 这样必须考虑幂等问题。
//                log.error("[消费者异常处理]，主题为：{}；分区为：{}；异常为：{}",
//                            mqTopic.topic,mqTopic.tag,e.getMessage());
//                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//            }
        });
        log.info("[消费者初始化完成], topic={}", prefix + mqTopic.topic);
        return consumer;
    }


    /**
     * 启动消费者 要在spring 容器启动后 进行一些监听 以遍可以从ioc容器中 拿到想要的bean，否则使用@PostConstruct 注解
     * 可能有一些bean 加载不到
     *
     * @param event
     */
    @EventListener(ApplicationPreparedEvent.class)
    public void startConsumer(ApplicationPreparedEvent event) {
        //获取bean
//        PushConsumerProperty property = event.getApplicationContext().getBean(PushConsumerProperty.class);
        try {
            DefaultMQPushConsumer consumer = initConsumer();
            consumer.setNamesrvAddr(namesrvAddr);
            consumer.start();
            //弃用 每个子类初始化的时候将子类里面的泛型值传到对应的父类进行一个初始化，这样父类就能直接调用子类传进来的泛型值了
            //弃用 每次只在初始化时候调用即可 不用每次都调用一次 有泛型擦除的问题
//            this.mClass = getActualTypeArguments();
            this.typeReference = getModel();


        } catch (MQClientException e) {
            log.error("[消费者start异常], nameServerAddress={}", namesrvAddr);
//            System.exit(-1);
        }
    }

//    @PreDestroy
//    public void closeConsumer() {
//        DefaultMQPushConsumer consumer = initConsumer();
//        consumer.shutdown();
//    }


    /**
     * 调用子类去消费消息
     *
     * @param msg
     */
    protected final void handlerMsg(MessageExt msg)  {
        String msgStr = new String(msg.getBody(), StandardCharsets.UTF_8);
        //new TypeReference 转类型 这里<String> 转完后就是String类型
//        M model= JSON.parseObject(msgStr, mClass);
        M model= JSON.parseObject(msgStr, typeReference);
        long storeTimestamp = msg.getStoreTimestamp();
        log.info("StoreTimestamp：{}；queueId：{}",getTime(storeTimestamp),msg.getQueueId());
        handler(model, msg.getKeys());
    }

    /**
     * 子类处理消息
     */
    protected abstract void handler(M model, String keys);


    /**
     * 父类获得子类相应的实体
     */
    protected abstract TypeReference<M> getModel();


    protected abstract MQTopic getMqTopic();


    /**
     * 父类接到子类传的泛型值
     */
    @SneakyThrows
    @Deprecated
    public Class<M> getActualTypeArguments() {
        // 若子类没有传泛型类型，则强转报错
        ParameterizedType genericSuperclass = (ParameterizedType) getClass().getGenericSuperclass();

        Type[] types = genericSuperclass.getActualTypeArguments();
        Class<M> tClass = (Class<M>) types[0];
//        Constructor<M> constructor = tClass.getDeclaredConstructor();
//        M myInstance = constructor.newInstance();
        return tClass;
    }

    /**
     * MQ的异常收集,存进去
     */
    private void errorConsumerCollect(MQTopic mqTopic, String msg, Exception e){
        if(!Objects.equals(MQTopic.ERROR.topic,mqTopic.topic)){
            log.info("进入ERROR的主题: {}",mqTopic.topic);
            MQError mqErrorParams = new MQError()
                    .setId(UUID.randomUUID().toString())
                    .setCrAt(new Date())
                    .setErrorMessage(e.getMessage())
                    .setErrorKeys(UUID.randomUUID().toString())
                    .setParamObjects(msg)
                    .setStatusFlag(0)
                    .setTag(mqTopic.tag)
                    .setTopic(mqTopic.topic)
                    .setTopicInfo(mqTopic.info)
                    .setType("consumer");
            try {
                errorProducer.send(MQTopic.ERROR, JSON.toJSONString(mqErrorParams));
            } catch (Exception ex) {
                log.error("转发失败：原因：{}",ex.getMessage());
            }
        }
    }


    private String getTime(long milliseconds){
        // 将毫秒数转换为 LocalDateTime 对象
        LocalDateTime dateTime = LocalDateTime.ofInstant(
                Instant.ofEpochMilli(milliseconds),
                ZoneId.systemDefault()
        );
        // 格式化日期时间
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        return dateTime.format(formatter);
    }

}
